/*
 * Copyright (c) 2023 Hofi <hofione@gmail.com>
 *
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License version 2 as published
 * by the Free Software Foundation, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 *
 * As an additional exemption you are allowed to compile & link against the
 * OpenSSL libraries as published by the OpenSSL project. See the file
 * COPYING for details.
 *
 */

#import "darwinosl-source.h"
#import "darwinosl-source-oslog.h"
#include "darwinosl-source-persist.h"
#include "logthrsource/logthrfetcherdrv.h"
#include "ack-tracker/ack_tracker_factory.h"
#include "stats/stats-cluster-single.h"
#include "stats/aggregator/stats-aggregator-registry.h"
#include "timeutils/misc.h"

/* TODO: Move these to darwinosl-source.h or a separate class once options and its handling need to be exposed */
void darwinosl_sd_options_defaults(DarwinOSLogSourceOptions *self,
                                   LogThreadedSourceWorkerOptions *super_source_options);
void darwinosl_sd_options_destroy(DarwinOSLogSourceOptions *self);
gboolean darwinosl_sd_options_get_read_old_records(DarwinOSLogSourceOptions *self);

struct _DarwinOSLogSourceOptions
{
  LogThreadedSourceWorkerOptions *super_source_options;
  MsgFormatOptions *format_options;

  gchar *filter_predicate;
  gboolean go_reverse;
  gboolean do_not_use_bookmark;
  guint max_bookmark_distance;
  guint fetch_delay;
  guint fetch_limit;
  gint fetch_retry_delay;
};

typedef struct _DarwinOSLogSourceDriver
{
  LogThreadedSourceDriver super;

  DarwinOSLogSourceOptions options;
  DarwinOSLogSourcePersist *log_source_persist;
  DarwinOSLogSourcePosition log_source_position;

  guint curr_fetch_in_run;
  const gchar *persist_name;
  const gchar *stat_persist_name;

  StatsAggregator *max_message_size;
  StatsAggregator *average_messages_size;
  StatsAggregator *CPS;

  /* NOTE: Naming convention difference is intentional here, camelCase is ObjC default.
           and signs nicely those variables need different kind of care because of ARC */
  __strong OSLogSource *osLogSource;
} DarwinOSLogSourceDriver;

static void
_log_reader_insert_msg_length_stats(DarwinOSLogSourceDriver *self, gsize len)
{
  stats_aggregator_add_data_point(self->max_message_size, len);
  stats_aggregator_add_data_point(self->average_messages_size, len);
}

static void
_register_aggregated_stats(DarwinOSLogSourceDriver *self)
{
  LogThreadedSourceWorker *worker = self->super.workers[0];
  StatsClusterKeyBuilder *kb = worker->super.metrics.stats_kb;
  stats_cluster_key_builder_push(kb);
  {
    LogSourceOptions *super_options = &self->options.super_source_options->super;
    gchar *stats_id = worker->super.stats_id;
    gchar stats_instance[1024];
    const gchar *instance_name = stats_cluster_key_builder_format_legacy_stats_instance(kb, stats_instance,
                                 sizeof(stats_instance));
    stats_aggregator_lock();
    StatsClusterKey sc_key;

    // msg_size_max
    stats_cluster_single_key_legacy_set_with_name(&sc_key,
                                                  super_options->stats_source | SCS_SOURCE,
                                                  stats_id,
                                                  instance_name, "msg_size_max");
    stats_register_aggregator_maximum(super_options->stats_level, &sc_key,
                                      &self->max_message_size);

    // msg_size_avg
    stats_cluster_single_key_legacy_set_with_name(&sc_key,
                                                  super_options->stats_source | SCS_SOURCE,
                                                  stats_id,
                                                  instance_name, "msg_size_avg");
    stats_register_aggregator_average(super_options->stats_level, &sc_key,
                                      &self->average_messages_size);

    // eps
    stats_cluster_single_key_legacy_set_with_name(&sc_key,
                                                  super_options->stats_source | SCS_SOURCE,
                                                  stats_id,
                                                  instance_name, "eps");
    stats_register_aggregator_cps(super_options->stats_level, &sc_key,
                                  worker->super.metrics.recvd_messages_key,
                                  SC_TYPE_SINGLE_VALUE,
                                  &self->CPS);
    stats_aggregator_unlock();
  }
  stats_cluster_key_builder_pop(kb);
}

static void
_unregister_aggregated_stats(DarwinOSLogSourceDriver *self)
{
  stats_aggregator_lock();

  stats_unregister_aggregator(&self->max_message_size);
  stats_unregister_aggregator(&self->average_messages_size);
  stats_unregister_aggregator(&self->CPS);

  stats_aggregator_unlock();
}

static void
_check_restored_postion(DarwinOSLogSourceDriver *self, OSLogEntry *nextLogEntry)
{
  NSString *msgString = [self->osLogSource stringFromDarwinOSLogEntry:nextLogEntry];

  if (g_str_hash(msgString.UTF8String) == self->log_source_position.last_msg_hash)
    {
      msg_debug("darwinosl: Bookmark is restored fine",
                evt_tag_str("bookmark", [OSLogSource.RFC3339DateFormatter stringFromDateWithMicroseconds:
                                                                          nextLogEntry.date].UTF8String));
    }
  else
    {
      if (self->log_source_position.last_msg_hash)
        msg_info("darwinosl: Could not restore last bookmark (filter might be changed or max_bookmark_distance took effect?)",
                 evt_tag_str("bookmark", [OSLogSource.RFC3339DateFormatter stringFromDateWithMicroseconds:
                                                                           [NSDate dateWithTimeIntervalSince1970:self->log_source_position.log_position]].UTF8String),
                 evt_tag_str("new_start_position", [OSLogSource.RFC3339DateFormatter stringFromDateWithMicroseconds:
                                                                                     nextLogEntry.date].UTF8String));
      else
        msg_debug("darwinosl: No last msg hash found",
                  evt_tag_str("new_start_position", [OSLogSource.RFC3339DateFormatter stringFromDateWithMicroseconds:
                                                                                      nextLogEntry.date].UTF8String));
    }
}

static gboolean
_log_position_date_from_persist(DarwinOSLogSourceDriver *self, NSDate **startDate)
{
  darwinosl_source_persist_load_position(self->log_source_persist, &self->log_source_position);

  /* NOTE: Here could be some internal logic to solve clashes might be caused by bookmarks saved with different bookmarking options, but
           based on read-old-records() we just let the user manually solve this (e.g. clear the persis-file if needed)
  */
  NSDate *maxBookmarkDistanceDate = [NSDate dateWithTimeIntervalSinceNow:-1 * (NSTimeInterval)(
                                              self->options.max_bookmark_distance)];
  if (self->options.max_bookmark_distance > 0)
    *startDate = maxBookmarkDistanceDate;

  gboolean should_start_from_now = (FALSE == darwinosl_sd_options_get_read_old_records(&self->options) &&
                                    (self->log_source_position.log_position == 0.0 || self->options.do_not_use_bookmark));
  if ((FALSE == self->options.do_not_use_bookmark && self->log_source_position.log_position != 0.0)
      || should_start_from_now)
    {
      NSDate *savedDate = [NSDate dateWithTimeIntervalSince1970:self->log_source_position.log_position];
      *startDate = (should_start_from_now ?
                    [NSDate dateWithTimeIntervalSinceNow:-1] :
                    (self->options.max_bookmark_distance > 0 ? [maxBookmarkDistanceDate laterDate:savedDate] : savedDate)
                   );
      return (FALSE == should_start_from_now);
    }
  return FALSE;
}

static gsize
_log_message_from_string(const char *msg_cstring, MsgFormatOptions *format_options, LogMessage **out_msg)
{
  gsize msg_len = strlen(msg_cstring);
  *out_msg = msg_format_construct_message(format_options, (const guchar *) msg_cstring, msg_len);
  msg_format_parse_into(format_options, *out_msg, (const guchar *) msg_cstring, msg_len);
  return msg_len;
}

static gboolean
_open_osl(DarwinOSLogSourceDriver *self)
{
  self->curr_fetch_in_run = 0;

  if ([self->osLogSource openStore] == nil)
    return FALSE;

  NSDate *startDate = nil;
  bool positionRestored = _log_position_date_from_persist(self, &startDate);
  NSString *filterString = (self->options.filter_predicate ? [NSString stringWithUTF8String:
                                                                       self->options.filter_predicate] : nil);
  OSLogEnumeratorOptions options = (self->options.go_reverse ? OSLogEnumeratorReverse : 0);

  msg_trace("darwinosl: Should start from",
            evt_tag_str("start_position",
                        (startDate ?
                         [OSLogSource.RFC3339DateFormatter stringFromDateWithMicroseconds:startDate].UTF8String :
                         "no saved position")));
  if ([self->osLogSource openEnumeratorWithDate:startDate filterString:filterString options:options] == nil)
    {
      [self->osLogSource closeAll];
      return FALSE;
    }

  if (positionRestored)
    {
      OSLogEntry *nextLogEntry = [self->osLogSource fetchNextEntry];
      if (nextLogEntry)
        _check_restored_postion(self, nextLogEntry);
      else
        msg_warning("darwinosl: Could not fetch log entry after a successful position restore");
    }

  self->log_source_position.last_used_filter_predicate_hash = (self->options.filter_predicate ?
                                                              g_str_hash(self->options.filter_predicate) :
                                                              0);
  return TRUE;
}

static inline void
_close_osl(DarwinOSLogSourceDriver *self)
{
  [self->osLogSource closeAll];
}

/* runs in a dedicated thread */
static LogThreadedFetchResult
_fetch(LogThreadedSourceWorker *worker)
{
  DarwinOSLogSourceDriver *self = (DarwinOSLogSourceDriver *) worker->control;
  gboolean fetchedEnough = (self->options.fetch_limit && self->curr_fetch_in_run >= self->options.fetch_limit);
  OSLogEntry *nextLogEntry = (fetchedEnough ? nil : [self->osLogSource fetchNextEntry]);

  if (nextLogEntry == nil)
    {
      msg_trace(fetchedEnough ? "darwinosl: Fetch limit reached" : "darwinosl: No more log data currently");
      msg_debug("darwinosl: Fetched logs in this run", evt_tag_long("fetch_count", (long) self->curr_fetch_in_run));

      log_threaded_source_worker_close_batch(worker);
      /*
         TODO: Dual return code is needed due to an actual macOS API bug https://openradar.appspot.com/radar?id=5597032077066240, as
               interrupting the reading from an open cursor would lead to unclosed enumerator threads with
               an unreleased OSLogEntry item array in each.
               Once the issue is fixed we can simplify this again to just use THREADED_FETCH_NO_DATA.
          So, now THREADED_FETCH_TRY_AGAIN means, continue reading (after iteration_sleep_time ellapsed)
              and THREADED_FETCH_NO_DATA means, close and reopen the store (after fetch_retry_delay elapsed)
        */
      LogThreadedFetchResult result = {fetchedEnough ? THREADED_FETCH_TRY_AGAIN : THREADED_FETCH_NO_DATA, NULL};
      /* this will not be needed either once the above mentioned issue is fixed */
      self->curr_fetch_in_run = 0;

      return result;
    }

  self->curr_fetch_in_run++;

  NSString *msgString = [self->osLogSource stringFromDarwinOSLogEntry:nextLogEntry];
  const char *log_string = msgString.UTF8String;
  msg_trace("darwinosl: Msg string is composed", evt_tag_str("msg", log_string));

  /* NOTE: No other options unfortunately (there is no "current position" like getter)
            If there are multiple occurrences of the same time, the method returns the earliest occurrence.
            So, in a worst case there might be log duplications on restart from bookmark
  */
  self->log_source_position.log_position = [nextLogEntry.date timeIntervalSince1970];
  self->log_source_position.last_msg_hash = g_str_hash(log_string);

  Bookmark *bookmark = ack_tracker_request_bookmark(worker->super.ack_tracker);
  darwinosl_source_persist_fill_bookmark(self->log_source_persist, bookmark, self->log_source_position);
  msg_trace("darwinosl: Bookmark created",
            evt_tag_printf("position", "%.6f", self->log_source_position.log_position),
            evt_tag_printf("last message hash", "%u", self->log_source_position.last_msg_hash),
            evt_tag_printf("filter predicate hash", "%u", self->log_source_position.last_used_filter_predicate_hash));

  LogMessage *msg;
  gsize msg_len = _log_message_from_string(log_string, self->options.format_options, &msg);
  log_msg_set_value_to_string(msg, LM_V_TRANSPORT, "local+darwinoslog");
  LogThreadedFetchResult result = {THREADED_FETCH_SUCCESS, msg};
  _log_reader_insert_msg_length_stats(self, msg_len);
  log_msg_set_recvd_rawmsg_size(msg, msgString.length);
  return result;
}

static inline void
_send(LogThreadedSourceWorker *worker, LogMessage *msg)
{
  log_threaded_source_worker_blocking_post(worker, msg);
}

/* runs in a dedicated thread */
static void
_run(LogThreadedSourceWorker *worker)
{
  DarwinOSLogSourceDriver *self = (DarwinOSLogSourceDriver *) worker->control;
  const gdouble iteration_sleep_time = self->options.fetch_delay ? 1.0 / self->options.fetch_delay : 0;
  gboolean exit_requested = FALSE;

  while (FALSE == (exit_requested = main_loop_worker_job_quit()))
    {
      @autoreleasepool
      {
        if (FALSE == _open_osl(self))
          break;

        while (FALSE == (exit_requested = main_loop_worker_job_quit()))
          {
            @autoreleasepool
            {
              LogThreadedFetchResult res = _fetch(worker);
              if (res.result == THREADED_FETCH_SUCCESS)
                _send(worker, res.msg);
              else if (res.result != THREADED_FETCH_TRY_AGAIN) /* See _fetch for meanng of THREADED_FETCH_TRY_AGAIN */
                break;
            }
            if (iteration_sleep_time > 0.0)
              main_loop_worker_wait_for_exit_until(iteration_sleep_time);
          }

        _close_osl(self);
      }
      if (FALSE == exit_requested && self->options.fetch_retry_delay)
        main_loop_worker_wait_for_exit_until(self->options.fetch_retry_delay);
    }
}

static const gchar *
_generate_persist_name(DarwinOSLogSourceDriver *self, const char *base_name, const char *name)
{
  gchar persist_name[1024];
  guint nameHash = 0;

  if (name == NULL && self->options.filter_predicate && strlen(self->options.filter_predicate))
    nameHash = g_str_hash(self->options.filter_predicate);

  const char *separator = (name || nameHash ? "." : "");
  if (nameHash)
    g_snprintf(persist_name, sizeof(persist_name), "%s%s%u", base_name, separator, nameHash);
  else
    g_snprintf(persist_name, sizeof(persist_name), "%s%s%s", base_name, separator, name ? : "");

  return g_strdup(persist_name);
}

static const gchar *
_get_persist_name(const LogPipe *s)
{
  DarwinOSLogSourceDriver *self = (DarwinOSLogSourceDriver *)s;

  if (self->persist_name == NULL)
    self->persist_name = _generate_persist_name(self, "darwinosl-source", s->persist_name);

  return self->persist_name;
}

static void
_format_stats_key(LogThreadedSourceDriver *s, StatsClusterKeyBuilder *kb)
{
  DarwinOSLogSourceDriver *self = (DarwinOSLogSourceDriver *)s;

  if (self->stat_persist_name == NULL)
    self->stat_persist_name = _generate_persist_name(self, "darwinosl", s->super.super.super.persist_name);
}

static gboolean
_init(LogPipe *s)
{
  DarwinOSLogSourceDriver *self = (DarwinOSLogSourceDriver *)s;
  GlobalConfig *cfg = log_pipe_get_config((LogPipe *)self);

  darwinosl_source_persist_init(self->log_source_persist, cfg->state, _get_persist_name(s));

  gboolean ret = log_threaded_source_driver_init_method(s);
  if (ret)
    _register_aggregated_stats(self);

  return ret;
}

static gboolean
_deinit(LogPipe *s)
{
  DarwinOSLogSourceDriver *self = (DarwinOSLogSourceDriver *)s;

  _unregister_aggregated_stats(self);

  return log_threaded_source_driver_deinit_method(s);
}

static void
_free(LogPipe *s)
{
  DarwinOSLogSourceDriver *self = (DarwinOSLogSourceDriver *)s;

  /* NOTE: ARC takes care of the rest */
  self->osLogSource = nil;

  darwinosl_sd_options_destroy(&self->options);
  darwinosl_source_persist_free(self->log_source_persist);

  if (self->persist_name)
    g_free((gpointer)self->persist_name);
  if (self->stat_persist_name)
    g_free((gpointer)self->stat_persist_name);

  log_threaded_source_driver_free_method(s);
}

static LogThreadedSourceWorker *
_construct_worker(LogThreadedSourceDriver *s, gint worker_index)
{
  /* TODO: DarwinOSLogSourceDriver uses the multi-worker API, but it is not prepared now to work with more than one worker. */
  g_assert(s->num_workers == 1);

  LogThreadedSourceWorker *worker = g_new0(LogThreadedSourceWorker, 1);
  log_threaded_source_worker_init_instance(worker, s, worker_index);

  worker->run = _run;

  return worker;
}

LogDriver *
darwinosl_sd_new(GlobalConfig *cfg)
{
  DarwinOSLogSourceDriver *self = g_new0(DarwinOSLogSourceDriver, 1);
  log_threaded_source_driver_init_instance(&self->super, cfg);

  self->super.super.super.super.init = _init;
  self->super.super.super.super.deinit = _deinit;
  self->super.super.super.super.free_fn = _free;
  self->super.super.super.super.generate_persist_name = _get_persist_name;

  self->super.worker_options.ack_tracker_factory = consecutive_ack_tracker_factory_new();
  self->super.worker_construct = _construct_worker;

  self->super.format_stats_key = _format_stats_key;

  self->log_source_persist = darwinosl_source_persist_new();

  self->osLogSource = [OSLogSource new];

  darwinosl_sd_options_defaults(&self->options, &self->super.worker_options);

  return &self->super.super.super;
}

/* Options */

/*
 * FIXME: Seems, in a debug build, there is no chance to catch the NSInvalidArgumentException, so
 *        this will always crash if the predicte string has certain parsing issues.
 */
gboolean
_try_test_filter_predicate_str(const gchar *filter_predicate_str)
{
  gboolean succ = TRUE;
  if (filter_predicate_str)
    {
      NSString *filterString = [NSString stringWithUTF8String:filter_predicate_str];
      if (filterString.length > 0)
        {
          NSPredicate *filterPredicate = nil;
          succ = FALSE;
          @try
            {
              if ((filterPredicate = [NSPredicate predicateWithFormat:filterString]) != nil)
                succ = TRUE;
            }
          @catch(NSException *e)
            {
              msg_error("darwinosl: Error in filter predicate",
                        evt_tag_str("predicate", filter_predicate_str),
                        evt_tag_str("error", [NSString stringWithFormat:@"%@", e].UTF8String));
            }
        }
    }
  return succ;
}

gboolean
darwinosl_sd_set_filter_predicate(LogDriver *s, const gchar *filter_predicate_str)
{
  DarwinOSLogSourceDriver *self = (DarwinOSLogSourceDriver *)s;

  if (self->options.filter_predicate)
    free (self->options.filter_predicate);
  self->options.filter_predicate = g_strdup(filter_predicate_str);

  if (self->options.filter_predicate)
    return _try_test_filter_predicate_str(self->options.filter_predicate);
  return TRUE;
}

void
darwinosl_sd_set_go_reverse(LogDriver *s, gboolean new_value)
{
  DarwinOSLogSourceDriver *self = (DarwinOSLogSourceDriver *)s;
  self->options.go_reverse = new_value;
}

void
darwinosl_sd_set_do_not_use_bookmark(LogDriver *s, gboolean new_value)
{
  DarwinOSLogSourceDriver *self = (DarwinOSLogSourceDriver *)s;
  self->options.do_not_use_bookmark = new_value;
}

void
darwinosl_sd_set_max_bookmark_distance(LogDriver *s, guint new_value)
{
  DarwinOSLogSourceDriver *self = (DarwinOSLogSourceDriver *)s;
  self->options.max_bookmark_distance = new_value;
}

void
darwinosl_sd_set_log_fetch_delay(LogDriver *s, guint new_value)
{
  DarwinOSLogSourceDriver *self = (DarwinOSLogSourceDriver *)s;
  self->options.fetch_delay = new_value;
}

void
darwinosl_sd_set_log_fetch_retry_delay(LogDriver *s, guint new_value)
{
  DarwinOSLogSourceDriver *self = (DarwinOSLogSourceDriver *)s;
  self->options.fetch_retry_delay = new_value;
}

void
darwinosl_sd_set_log_fetch_limit(LogDriver *s, guint new_value)
{
  DarwinOSLogSourceDriver *self = (DarwinOSLogSourceDriver *)s;
  self->options.fetch_limit = new_value;

  // We use our batch handling if fetch_limit is set
  self->super.auto_close_batches = (self->options.fetch_limit == 0);
}

gboolean
darwinosl_sd_options_get_read_old_records(DarwinOSLogSourceOptions *self)
{
  return self->super_source_options->super.read_old_records;
}

void
darwinosl_sd_options_defaults(DarwinOSLogSourceOptions *self,
                              LogThreadedSourceWorkerOptions *super_source_options)
{
  self->super_source_options = super_source_options;
  self->super_source_options->super.stats_level = STATS_LEVEL0;
  self->super_source_options->super.stats_source = stats_register_type("darwinosl");
  self->super_source_options->super.read_old_records = FALSE;

  /* No additional format options now, so intentionally referencing only, and not using msg_format_options_copy */
  self->format_options = &self->super_source_options->parse_options;

  self->filter_predicate = NULL;
  self->go_reverse = FALSE;
  self->do_not_use_bookmark = FALSE;
  self->max_bookmark_distance = 0; /* 0 means no limit */
  self->fetch_delay = 10000;
  self->fetch_limit = 0; /* 0 means no limit */
  self->fetch_retry_delay = 1; /* seconds */
}

void
darwinosl_sd_options_destroy(DarwinOSLogSourceOptions *self)
{
  if (self->filter_predicate)
    {
      free (self->filter_predicate);
      self->filter_predicate = NULL;
    }

  /* Just referenced, not copied */
  self->format_options = NULL;
  self->super_source_options = NULL;
}
